package com.xrui.hbase.codec.impl;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.protobuf.AbstractMessage;
import com.google.protobuf.Descriptors.Descriptor;

import com.xrui.hbase.DecodedCell;
import com.xrui.hbase.codec.CellEncoder;
import com.xrui.hbase.schema.SchemaType;
import com.xrui.hbase.schema.impl.ProtobufCellSchema;
import com.xrui.hbase.schema.spec.CellSpec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;

/**
 * Serializes Protocol Buffers for persistence into an HBase cell.
 * <p>
 * <p>
 * A Protocol Buffer cell encoder is specific to one column in a DatabaseTable.
 * Protocol Buffers generated classes know how to encode and decode themselves.
 * <p>
 */
public final class ProtobufCellEncoder implements CellEncoder {
    /* The name of this provider */
    public static final String PROVIDER_NAME = "ProtobufCellEncoder";
    private static final Logger LOG = LoggerFactory.getLogger(ProtobufCellEncoder.class);

    /**
     * Specification of the column encoding.
     */
    private final CellSpec mCellSpec;

    /**
     * Full-name of the protocol buffer to encode.
     */
    private final String mProtobufFullName;

    /**
     * Class of the protocol buffer to encode.
     */
    private final Class<?> mProtoClass;

    /**
     * Creates a new <code>CellEncoder</code> instance.
     *
     * @param cellSpec Specification of the cell to encode.
     * @throws IOException on I/O error.
     */
    public ProtobufCellEncoder(final CellSpec cellSpec) throws IOException {
        mCellSpec = Preconditions.checkNotNull(cellSpec);
        Preconditions.checkArgument(cellSpec.getCellSchema().getType() == SchemaType.PROTOBUF);
        ProtobufCellSchema cellSchema = (ProtobufCellSchema) cellSpec.getCellSchema();
        mProtobufFullName = cellSchema.getProtobufFullName();
        Preconditions.checkNotNull(mProtobufFullName);

        final String className = cellSchema.getProtobufClassName();
        Class<?> protoClass = null;
        try {
            protoClass = Class.forName(className);
            final Descriptor descriptor =
                (Descriptor) protoClass.getMethod("getDescriptor").invoke(protoClass);
            if (!Objects.equal(descriptor.getFullName(), mProtobufFullName)) {
                throw new IOException(String.format(
                    "Protocol buffer from class %s has full name %s, "
                        + "and does not match expected protocol buffer name %s.",
                    className, descriptor.getFullName(), mProtobufFullName));
            }
        } catch (NoSuchMethodException nsme) {
            LOG.error("Unable to find protocol buffer method : {}.getDescriptor()", className);
            protoClass = null;
        } catch (IllegalAccessException iae) {
            LOG.error("Illegal access on protocol buffer method : {}.getDescriptor(): {}",
                className, iae);
            protoClass = null;
        } catch (InvocationTargetException ite) {
            LOG.error("Invocation target exception on protocol buffer method : {}.getDescriptor(): {}",
                className, ite);
            protoClass = null;
        } catch (ClassNotFoundException cnfe) {
            LOG.error("Unable to load protocol buffer class: {}", className);
        }
        mProtoClass = protoClass;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public byte[] encode(final DecodedCell<?> cell) throws IOException {
        return encode(cell.getData());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized <T> byte[] encode(final T cellValue) throws IOException {
        final AbstractMessage message = (AbstractMessage) cellValue;
        if (mProtoClass != message.getClass()) {
            throw new IOException(String.format(
                "Protocol buffer of class '%s' (message name '%s') "
                    + "does not match expected class '%s' (message name '%s'): %s.",
                cellValue.getClass().getName(), message.getDescriptorForType().getFullName(),
                mProtoClass.getName(), mProtobufFullName,
                cellValue));
        }
        return message.toByteArray();
    }

    @Override
    public String getName() {
        return PROVIDER_NAME;
    }
}
